package com.atguigu.sql.connector;
/*
*
* DataGenConnector:用于模拟生成数据
*
* PrintConnector:用于输出
*
* */

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.stream.Stream;

public class Flink01_DataGenAndPrint {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //表环境
        //TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build());

        StreamTableEnvironment streamTableEnv = StreamTableEnvironment.create(env);

        //DataGen:模拟生成数据源
        String sourceTable =
                "create table t_source ( " +
                        "id STRING," +
                        "vc INT," +
                        "ts BIGINT " +
                " ) WITH ( " +
                        " 'connector' = 'datagen'," +
                        " 'rows-per-second' = '1'," +
                        " 'number-of-rows' = '1000'," +
                        " 'fields.id.kind'='random'," +
                        " 'fields.id.length'= '6', " +
                        " 'fields.vc.kind' = 'random'," +
                        " 'fields.vc.min' = '500'," +
                        " 'fields.vc.max'= '1000'," +
                        " 'fields.ts.kind' = 'sequence'," +
                        " 'fields.ts.start' = '10000' , " +
                        " 'fields.ts.end' = '99999'" +
                 " ) ";
        streamTableEnv.executeSql(sourceTable);

        //查询数据
        Table resultTable = streamTableEnv.sqlQuery(" select id,vc,ts from t_source where vc >=700");
        streamTableEnv.createTemporaryView("result_table",resultTable);
        //将表的结果输出到控制台
        String sinkTable =
                "create table t_sink ("+
                        "id STRING,"+
                        "vc INT," +
                        "ts BIGINT" +
                ") WITH (" +
                        "'connector' = 'print'," +
                        "'print-identifier' = 'PT'," +
                        "'standard-error' = 'false'," +
                        "'sink.parallelism' = '1'" +
                ")";
        streamTableEnv.executeSql(sinkTable);

        //写出
        //streamTableEnv.executeSql("insert into t_sink select id,vc,ts from result_table");
        resultTable.executeInsert("t_sink");


    }
}
